package ex.datastream.connectors;

import ex.datatypes.TaxiRide;
import ex.util.generator.TaxiRideGenerator;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * 将自动生成的数据存储到mysql数据库
 */
public class ToMysqlTest {
    private final SourceFunction<TaxiRide>source;
    private final SinkFunction<TaxiRide> sink;

    public ToMysqlTest(SourceFunction<TaxiRide> source, SinkFunction<TaxiRide> sink) {
        this.source = source;
        this.sink = sink;
    }

    public static void main(String[] args) throws Exception {
        //自动生成数据
        ToMysqlTest mysql=new ToMysqlTest(new TaxiRideGenerator(), new PrintSinkFunction<>());
        mysql.execute();
    }

    public JobExecutionResult execute() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //1、输入(自动生成数据)
        DataStream<TaxiRide> dataStream=env.addSource(source);
        Table inputTable = tableEnv.fromDataStream(dataStream);

        //2、输出(mysql)
        tableEnv.executeSql("CREATE TABLE taxi_ride (" +
                "    rideId BIGINT," +
                "    isStart    BOOLEAN," +
                "    eventTime    TIMESTAMP(3)," +
                "    startLon    FLOAT," +
                "    startLat    FLOAT," +
                "    endLon    FLOAT," +
                "    endLat    FLOAT," +
                "    passengerCnt    SMALLINT," +
                "    taxiId    BIGINT," +
                "    driverId     BIGINT," +
                "    PRIMARY KEY (rideId, driverId) NOT ENFORCED" +
                ") WITH (" +
                "   'connector'  = 'jdbc'," +
                "   'url'        = 'jdbc:mysql://localhost:3306/test?characterEncoding=utf8&serverTimezone=Asia/Shanghai&useSSL=false'," +
                "   'table-name' = 'taxi_ride'," +
                "   'driver'     = 'com.mysql.cj.jdbc.Driver'," +
                "   'username'   = 'root'," +
                "   'password'   = 'liji@2024'," +
                "   'sink.buffer-flush.max-rows' = '0'" +
                ")");

        //3、查询临时视图数据
        Table result=inputTable.select($("*"));


        //4、临时视图数据存入mysql数据库中
        result.insertInto("taxi_ride").execute();
        result.printSchema();

        dataStream.addSink(sink);
        return env.execute();
    }
}
